#include <common/macro/debug.h>

#include <database/database.h>
#include <transaction/transaction.h>
#include <storage/page.h>
#include <storage/storage.h>
#include <storage/blob/page_blob.h>
#include <storage/tuple/tuple.h>
#include <storage/tuple/page_tuple.h>

#include <maintenance/vacuum.h>

/** 
* @brief Vacuum the given storage.
* 
* @param transaction
* @param storage
* @param last_finished
* 
* @return Zero on success, -1 on error.
*/
API_FILELOCAL int maintenance_vacuum_table_internal(transaction_t *transaction, storage_t *storage, transaction_id_t last_finished) {
	int page_status, fsm_status;
	int page_dirty, fsm_dirty;
	page_number_t page_number;
	page_size_t page_size;
	page_item_t page_item;
	page_type_e page_type;
	transaction_phase_e phase;
	page_t *page;
	tuple_t *tuple;

	page_number = 0;
	while (1) {
		page = NULL;
		page_dirty = 0;
		fsm_dirty = 0;
		
		if (storage_page_vacuumlock(storage, page_number) != 0)
			return -1;
		if (storage_page_writelock(storage, page_number) != 0) {
			storage_page_vacuumunlock(storage, page_number);
			return -1;
		}

		page_status = storage_page_read(storage, page_number, &page);
		fsm_status = storage_fsm_get_info(storage, page_number, &page_type, &page_size);

		if (page_status != 0 && fsm_status != 0) {
			/* Assuming EOF: Neither page nor FSM is readable. */
			break;
		} else if (page_status != 0 && fsm_status == 0) {
			/* ERROR - unparsable page, with FSM entry. */
			/*
			 * TODO:
			 * a) dump data from the page into lost+found
			 * b) rewrite page with an empty PAGE_ANY page
			 * c) notify FSM about it.
			 */
			continue;
		} else if (page_status == 0 && fsm_status != 0) {
			/* ERROR - page was succesfully read, but lacks valid FSM entry. */
		}

		switch (page->type) {
			case PAGE_TUPLE:
				page_item = 0;
				while (page_item < page->tuple->tuples_c) {
					tuple = &page->tuple->tuples[page_item];
	
					transaction_visibility(transaction, tuple->id_insert, &phase);
					if (phase == ABORTED) {
						/* The tuple insertion was aborted. We can remove it. */
						page_tuple_delete(page->tuple, page_item);
						page_dirty = 1;
						continue;
					}
	
					if (tuple->deleted) {
						/* If the tuple was released, physically remove it. */
						transaction_visibility(transaction, tuple->id_delete, &phase);
						if (phase == COMMITED) {
							page_tuple_delete(page->tuple, page_item);
							page_dirty = 1;
							continue;
						}

						/*
						 * A specific case of a mid-transaction vacuum. Is it a tuple
						 * this transaction has both created and deleted?
						 */
						if (tuple->id_insert == tuple->id_delete && tuple->id_delete == transaction->id) {
							page_tuple_delete(page->tuple, page_item);
							page_dirty = 1;
							continue;
						}
					}
					
	
					if (transaction_list_compare_ids(transaction->transaction_list,
				                                     tuple->id_insert, last_finished) < 0) {
						tuple->id_insert = last_finished;
						page_dirty = 1;
					}

					if (transaction_list_compare_ids(transaction->transaction_list,
				                                     tuple->id_origin, last_finished) < 0) {
						tuple->id_origin = last_finished;
						page_dirty = 1;
					}
					page_item++;
				}

				/* Did we actually empty the page? */
				if (page->tuple->tuples_c == 0) {
					page_set_empty(page);
					page_dirty = 1;
				}
				break;

			case PAGE_BLOB:
				transaction_visibility(transaction, page->blob->id_insert, &phase);
				if (phase == ABORTED) {
					/* The tuple insertion was aborted. We can remove it. */
					page_set_empty(page);
					page_dirty = 1;
					break;
				}
	
				if (page->blob->deleted) {
					/* If the tuple was released, physically remove it. */
					transaction_visibility(transaction, page->blob->id_delete, &phase);
					if (phase == COMMITED) {
						page_set_empty(page);
						page_dirty = 1;
						break;
					}
				}
	
				if (transaction_list_compare_ids(transaction->transaction_list,
				                                 page->blob->id_insert, last_finished) < 0) {
					page->blob->id_insert = last_finished;
					page_dirty = 1;
				}
				break;

			default:
				break;
		}

		/*
		 * The page is authoritative source of information, not FSM.
		 */
		if (page->type != page_type) {
			page_type = page->type;
			fsm_dirty = 1;
		}
		if (page_get_freespace(page) != page_size) {
			page_size = page_get_freespace(page);
			fsm_dirty = 1;
		}

		if (fsm_dirty) {
			storage_fsm_update(storage, page_number, page_type, page_size);
		}

		if (page_dirty) {
			storage_page_write(storage, page_number, page);
		}
		storage_fsm_unlock(storage, page_number, 1);
		storage_page_rwunlock(storage, page_number);
		storage_page_vacuumunlock(storage, page_number);

		page_destroy(page);
		page = NULL;

		page_number++;
	}

	return 0;
}

/** 
* @brief Vacuum all tables in the database, using given transaction 
* reference.
* 
* @param transaction
* 
* @return Zero on success, -1 on error.
*/
API_PUBLIC int maintenance_vacuum_database(transaction_t *transaction) {
	int i, rv;
	transaction_id_t last_finished;

	transaction_refresh(transaction);

	transaction_log_writelock_vacuum(transaction->database->transaction_log);
	last_finished = transaction_list_last_finished(transaction->transaction_list);

	database_table_open_all(transaction->database);
	for (i = 0; i < transaction->database->tables_c; i++) {
		rv = maintenance_vacuum_table_internal(transaction, transaction->database->tables[i].storage, last_finished);
		if (rv != 0) {
			transaction_log_unlock_vacuum(transaction->database->transaction_log);
			return rv;
		}
	}

	abort_log_clear(transaction->database->abort_log);
	abort_log_flush(transaction->database->abort_log);

	transaction_log_set_vacuum(transaction->database->transaction_log, last_finished);
	transaction_log_unlock_vacuum(transaction->database->transaction_log);
	return 0;
}

/** 
* @brief Wrapper function, allowing vacuuming of only one table.
* 
* @param transaction
* 
* @return Zero on success, -1 on error.
*/
API_PUBLIC int maintenance_vacuum_table(transaction_t *transaction, const char *tablename) {
	int rv;
	table_storage_t table_storage;
	transaction_id_t last_finished;

	last_finished = transaction_list_last_finished(transaction->transaction_list);
	rv = database_table_open(transaction->database, tablename, &table_storage);
	if (rv != 0)
		return rv;
	
	return maintenance_vacuum_table_internal(transaction, table_storage.storage, last_finished);
}
